package com.xinqing.test.flink.datastream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author:CHQ
 * @Date:2021/4/20 10:01
 * @Description
 */
public class CheckPoint {
    public static void main(String args[]) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //每隔5s钟记录一次checkpoint，模式为精准一次
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

        //规定每一次的checkpoint必须在1min之内完成，否则被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        //checkpoint并行数量为1
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);


        List<String> list = new ArrayList<>();
        list.add("1");
        list.add("2");
        list.add("3");

        env.fromCollection(list).map((MapFunction<String, Object>) value -> Integer.parseInt(value) + 1).print();

        env.execute();
    }
}
